package airdata

import (
	"context"
	"fmt"
	"github.com/robfig/cron/v3"
	"iot-base/auth/db"
	"iot-base/auth/logger"
	"iot-base/auth/svc"
	"iot-base/common/cache"
	"iot-base/common/params"
	"iot-base/common/redis"
	"strings"
	"sync"
	"time"
)

type AirData struct {
	Value     float64
	Status    int //@ 状态 0 未知 1 正常  2 校准  3 开机 4 中断恢复 5 诊断 6 零点校准 7 跨度校准 8 有效数据不足
	CreatTime time.Time
	Unit      string
}

//键值存储当前项目co,no,no2,nox等,值为当前采集的数据
type chanDt chan map[string]AirData

var (
	m          = sync.Mutex{}
	CronManger map[string]context.CancelFunc
	modelStr   = []string{"m_co", "m_nox", "m_o3", "m_pm25", "m_pm10", "m_so2", "m_atmosphere"}
	reportList = map[string][]string{
		//键值为物模型id，内容为需要上报的项目名称与物模型属性的转换规则自定义
		"m_pm25":       {"PM2.5"},
		"m_pm10":       {"PM10"},
		"m_co":         {"CO"},
		"m_nox":        {"NO", "NO2", "NOX"},
		"m_so2":        {"SO2"},
		"m_o3":         {"O3"},
		"m_atmosphere": {"wind speed", "wind direction", "temperature", "pressure", "humity", "rainfall total", "rainfall hour", "ranfall minute"},
	}
	//atmosphereMap = map[string]string{
	//	"wind speed":     "风速",
	//	"wind direction": "风向",
	//	"temperature":    "气温",
	//	"pressure":       "气压",
	//	"humity":         "湿度",
	//	"ranfall minute": "雨量",
	//	"rainfall hour":  "雨量",
	//	"rainfall total": "雨量",
	//}
	//此模块用于获取当前需要的上报数据的查询
	rc              = db.NewRedis()
	consumerReal    map[int64]chanDt
	consumer5minute map[int64]chanDt
	consumerHour    map[int64]chanDt

	consumer5minuteDev map[int64]chanDevDt
	consumerHourDev    map[int64]chanDevDt
	Cron               = cron.New(cron.WithParser(cron.NewParser(cron.Second|cron.Minute|
		cron.Hour|cron.Dom|cron.Month|cron.DowOptional|cron.Descriptor)), cron.WithChain())
)

func init() {
	CronManger = make(map[string]context.CancelFunc)
	consumerReal = make(map[int64]chanDt)
	consumer5minute = make(map[int64]chanDt)
	consumerHour = make(map[int64]chanDt)

	consumer5minuteDev = make(map[int64]chanDevDt)
	consumerHourDev = make(map[int64]chanDevDt)
	Cron.Start()
	_, err := Cron.AddFunc("0 3/5 * * * *", get5minuteData)
	_, err = Cron.AddFunc("0 3/5 * * * *", get5minuteDev)
	if err != nil {
		logger.Log.Error("5 minute timer is err")
	}
	_, err = Cron.AddFunc("0 0 */1 * * *", getHourData)
	_, err = Cron.AddFunc("0 0 */1 * * *", getHourDev)
	if err != nil {
		logger.Log.Error("1 hour timer is err")
	}
	_, err = Cron.AddFunc("*/15 * * * * *", getRealData)
	if err != nil {
		logger.Log.Error("real timer is err")
	}
}

func getRealData() {
	//fmt.Println("goroute count:", runtime.NumGoroutine())
	devs, err := svc.Dao.TThings.QueryByModelId(modelStr)
	if err != nil {
		logger.Log.Error("db err")
		return
	}
	var result = make(map[string]AirData)
	for _, dev := range devs {
		view, err := cache.GetThingView(dev.UID)
		if err != nil {
			logger.Log.Error("get redis cache err id is " + dev.UID)
			return
		}
		var temValue = make(map[string]params.ValueInRedis)
		var status int
		for _, v := range view.ValuesPtr {
			for _, pro := range reportList[dev.ModelID] {
				if v.PID == strings.ToLower(pro) {
					if v.ModelID != "m_atmosphere" {
						temValue[strings.ToLower(pro)] = *v
					}
				}
			}
			if v.PID == "status" { // # 状态 0 未知 1 正常  2 校准  3 开机 4 中断恢复 5 诊断 6 零点校准 7 跨度校准
				status = int(v.IValue)
			}
			if status == 2 {
				data, err := rc.HGetAll(redis.CaliInfoKey)
				if err != nil {
					logger.Log.Error("hget redis data err!", err)
				} else {
					modelid := data["model_id"]
					iscali := data["is_cali"]
					if modelid == v.ModelID && iscali == "1" {
						gastype := data["gas_type"]
						if gastype == "0" {
							status = 6
						} else if gastype == "1" {
							status = 7
						}
					}
				}
			}
		}
		for k, v := range temValue {
			creat := time.Unix(v.PickTime, 0)
			if time.Now().Add(-20*time.Second).Unix() < v.PickTime {
				result[k] = AirData{
					Value:     v.FValue,
					Status:    status,
					CreatTime: creat,
					Unit:      v.UnitName,
				}
			}
		}
	}
	//发布到每一个订阅实时数据
	for _, v := range consumerReal {
		v <- result
	}
}

func get5minuteData() {
	hisdata, err := svc.Dao.THisData.QueryHisByModelTimeStyle(modelStr, "5minute", time.Now().Add(-11*time.Minute))
	if err != nil {
		return
	}
	var result = make(map[string]AirData)
	for _, v := range hisdata {
		var airData = AirData{
			Value:     v.Value,
			Status:    v.Status,
			CreatTime: v.PickTime,
			Unit:      v.Unit,
		}
		for k, properties := range reportList {
			if k == v.ModelID {
				for _, property := range properties {
					if strings.ToUpper(v.Field) == property || v.Field == property {
						if v.ModelID == "m_atmosphere" {
							_, item := cache.GetModelProperty(v.ModelID, v.Field, "")
							result[item.Name] = airData
						} else {
							result[property] = airData
						}
					} else if v.Field == "pm" {
						switch v.ModelID {
						case "m_pm25":
							result["pm2.5"] = airData
						case "m_pm10":
							result["pm10"] = airData
						}
					}
				}
			}
		}
	}
	for _, v := range consumer5minute {
		v <- result
	}
}

func getHourData() {
	hisdata, err := svc.Dao.THisData.QueryHisByModelTimeStyle(modelStr, "hour", time.Now().Add(-1*time.Hour-1*time.Minute))
	if err != nil {
		return
	}
	var result = make(map[string]AirData)
	for _, v := range hisdata {
		var airData = AirData{
			Value:     v.Value,
			Status:    v.Status,
			CreatTime: v.PickTime,
			Unit:      v.Unit,
		}
		if v.Total < 9 {
			airData.Status = 8
		}
		for k, properties := range reportList {
			if k == v.ModelID {
				for _, property := range properties {
					if strings.ToUpper(v.Field) == property || v.Field == property {
						if v.ModelID == "m_atmosphere" {
							_, item := cache.GetModelProperty(v.ModelID, v.Field, "")
							result[item.Name] = airData
						} else {
							result[property] = airData
						}
					} else if v.Field == "pm" {
						switch v.ModelID {
						case "m_pm25":
							result["pm2.5"] = airData
						case "m_pm10":
							result["pm10"] = airData
						}
					}
				}
			}
		}
	}
	for _, v := range consumerHour {
		v <- result
	}
}

//@id 用于注册的id
//@style 用于区分订阅的数据 real 5minute hour
//@return 数据通过返回通道进行传输，map中键值为项目(pm2.5,pm10,so2,o3,co,no,no2,nox)
func subscribeAirData(id int64, style string) chan map[string]AirData {
	m.Lock()
	defer m.Unlock()
	switch style {
	case "real":
		consumerReal[id] = make(chan map[string]AirData, 10)
		return consumerReal[id]
	case "5minute":
		consumer5minute[id] = make(chan map[string]AirData, 10)
		return consumer5minute[id]
	case "hour":
		consumerHour[id] = make(chan map[string]AirData, 10)
		return consumerHour[id]
	}
	return nil
}

func SubscribeByFunc(id int64, style string, run func(map[string]AirData) error) {
	data := subscribeAirData(id, style)
	ctx, cancel := context.WithCancel(context.Background())
	m.Lock()
	CronManger[fmt.Sprint(id)+"."+style] = cancel
	m.Unlock()
	go func(c context.Context) {
		for {
			select {
			case inData := <-data:
				err := run(inData)
				if err == nil {
					fmt.Println("err:", err)
				}
				//fmt.Println("deal data end")
			case <-c.Done():
				return
			}
		}
	}(ctx)
}

func DisSubscribeAirData(id int64, style string) {
	m.Lock()
	defer m.Unlock()
	if _, ok := CronManger[fmt.Sprint(id)+"."+style]; ok {
		CronManger[fmt.Sprint(id)+"."+style]()
	}
	switch style {
	case "real":
		delete(consumerReal, id)
	case "5minute":
		delete(consumer5minute, id)
	case "hour":
		delete(consumerHour, id)
	case "all":
		delete(consumerReal, id)
		delete(consumer5minute, id)
		delete(consumerHour, id)
	}
}
